package com.intct.flink.study;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-10-11 16:57
 */
public class SQLTest01 {
    public static void main(String[] args) {
        // 配置属性
        Configuration conf = new Configuration();
        conf.set(RestOptions.BIND_PORT, "8081");

        // 1 获取取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        // 并行度
        // env.setParallelism(1);
        // 开启检点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 获取SQL执行环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        tenv.executeSql(
                "SELECT\n" +
                        " supplier_id\n" +
                        " , rating\n" +
                        " , product_id\n" +
                        " , COUNT(*)\n" +
                        " FROM (\n" +
                        " VALUES\n" +
                        " ('supplier1', 'product1', 4),\n" +
                        " ('supplier1', 'product2', 3),\n" +
                        " ('supplier2', 'product3', 3),\n" +
                        " ('supplier2', 'product4', 4)\n" +
                        " )\n" +
                        "AS Products(supplier_id, product_id, rating)\n" +
                        " GROUP BY GROUPING SETS(\n" +
                        " (supplier_id, product_id, rating),\n" +
                        " (supplier_id, product_id),\n" +
                        " (supplier_id, rating),\n" +
                        " (supplier_id),\n" +
                        " (product_id, rating),\n" +
                        " (product_id),\n" +
                        " (rating),\n" +
                        " ()\n" +
                        " )").print();
    }
}
